package com.xian.spark.wordcount;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

/**
 * 本地版的spark wordcount
 */
public class MyWordCountLocal {
    public static void main(String[] args) {
        //设置配置
        SparkConf sparkConf = new SparkConf();
        sparkConf.setAppName("java wordcount local");
        sparkConf.setMaster("local");

        JavaSparkContext jsc = new JavaSparkContext(sparkConf);
        JavaRDD<String> fileRDD = jsc.textFile("e:/test/wc.txt");
        JavaRDD<String> wordsRDD = fileRDD.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String line) throws Exception { //遍历文件每一行
                System.out.println("当前内容:" + line);
                String[] wordArr = line.split(" "); //每一行切割出的多个词数组
                List<String> wordList = Arrays.asList(wordArr);
                return wordList.iterator();
            }
        });

        JavaPairRDD<String, Long> word_one_RDD = wordsRDD.mapToPair(new PairFunction<String, String, Long>() {
            @Override
            public Tuple2<String, Long> call(String word) throws Exception {
                Tuple2<String, Long> word_one = new Tuple2<>(word, 1L);
                return word_one;
            }
        });

        JavaPairRDD<String, Long> word_count_RDD = word_one_RDD.reduceByKey(new Function2<Long, Long, Long>() {
            @Override
            public Long call(Long v1, Long v2) throws Exception {
                return v1 + v2;
            }
        });


        word_count_RDD.foreach(new VoidFunction<Tuple2<String, Long>>() {
            @Override
            public void call(Tuple2<String, Long> word_count) throws Exception {
                System.out.println(word_count._1 + "==" + word_count._2());
            }
        });



        jsc.stop();
        jsc.close();






    }
}
